package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.func.DimAsyncFunction;
import com.atguigu.gmall.realtime.beans.TradeSkuOrderBean;
import com.atguigu.gmall.realtime.utils.DateFormatUtil;
import com.atguigu.gmall.realtime.utils.MyClickhouseUtil;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import com.atguigu.gmall.realtime.utils.TimestampLtz3CompareUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

/**
 * @author Felix
 * @date 2022/11/11
 * 交易域：sku粒度下单业务过程聚合统计
 * 需要启动的进程
 *      zk、kafka、maxwell、hdfs、hbase、Redis、Clickhouse、DwdTradeOrderPreProcess、DwdTradeOrderDetail
 *      DwsTradeSkuOrderWindow
 * 开发流程
 *      基本环境准备
 *      检查点相关的设置
 *      从kafka的下单主题中读取数据
 *      将读取到的json字符串转换为json对象
 *      去重
 *          为什么会出现重复？
 *              因为我们是从下单事实表中读取数据，下单事实表数据来源于订单预处理表，订单预处理表是将订单、订单明细、
 *              订单明细活动、订单明细优惠券、字典表5张表关联在一天。在和明细活动以及明细优惠券关联的时候，使用的是左外连接，
 *              如果左表数据先到，右表数据后到，左外连接会出现如下数据
 *              +I 左表   null
 *              -D 左表   null
 *              +I 左表   右表
 *              将上述的三条数据写到kafka主题中会得到3条结果
 *                  左表  null
 *                  null
 *                  左表  右表
 *              因为我们在开发dwd的下单应用的时候，使用的是kafka连接器从预处理主题中读取数据，所以null会被自动去掉，但是存在左表数据重复
 *          去重思路？
 *              Flink状态编程 + 定时器
 *              按照唯一键进行分组(唯一键是订单明细的id)
 *              对分组之后的数据进行处理，当组内的数据来的时候，判断状态中是否存在数据，如果状态中不存在数据，说明是第一条数据，
 *              将这条数据放到状态中，并注册5秒钟之后执行的定时器；
 *              如果当组内数来的时候，状态中已经存在了历史数据，用状态中数据的时间和当前这条来的数据的时间进行比较，将时间大的
 *              保留到状态中；等5秒之后的定时器执行了，将状态中的数据写到下游，并清除状态
 *      类型转换
 *          jsonObj->实体类对象
 *      按照user_id进行分组
 *      使用Flink的状态编程，判断是否为下单独立用户
 *      指定Watermark以及提取事件时间字段
 *      按照统计的维度sku_id进行分组
 *      开窗
 *      聚合计算
 *      和维度进行关联
 *          基本维度关联
 *              PhoenixUtil
 *                  List<T> queryList(String sql,Class<T> clz)
 *              DimUtil
 *                  JSONObject getDimInfoNoCache(String tableName,Tuple2...params)
 *          优化1：旁路缓存
 *              思路：先从缓存中查询维度数据，如果缓存中存在要查询的维度，直接作为返回值返回(缓存命中);如果缓存中不存在要查询的维度，
 *                  再发送请求到phoenix表中将维度查询出来，并将查询出来的维度放到缓存进行缓存，方便下次查询
 *               缓存产品选型
 *                  状态:     性能好、维护性差
 *                  Redis：  性能也不错、维护性号  √
 *               Redis相关配置:
 *                  key:    dim:维度表表名:主键1_主键2
 *                  type:   String
 *                  expire: 1day
 *               注意：业务数据库维度表发生了变化-->phoenix对应的维度表-->删除Redis中缓存的维度数据
 *
 *          优化2：异步IO
 *              同步： 排队执行
 *              异步： 同时执行
 *              为什么在和外部数据库进行维度关联的时候，要使用异步操作？
 *                  在Flink程序中，如果使用map算子对流中数据进行处理，默认使用的是同步的方式，也就是说，处理完流中的一个元素后，再
 *              去处理下一个元素，这样的话，如果和外部系统进行交互的话，等待时间就过长，时效性较差，虽然可以加大并行度，提升一下处理
 *              能力，但是更多的并行度意味着更多的硬件资源，不可能无限制的提升。
 *              所以我们考虑在和外部系统交互的时候，使用异步的方式，所以异步，就是同时处理流中的多个元素
 *              Flink提供了专门的API进行异步操作
 *                  AsyncDataStream.[un]orderedWait(
 *                      流,
 *                      发送异步请求 implements AsyncFunction，
 *                      超时时间,
 *                      时间单位
 *                  )
 *                  class DimAsyncFunction extends RichAsyncFunction{
 *                      open{
 *                          初始化线程池对象
 *                      }
 *                      asyncInvoke{
 *                          开启新的线程，发送异步请求
 *                          模板方法设计模式：在当前类中定义完成维度关联的核心步骤，但是具体的实现延迟到子类中去完成
 *                      }
 *                  }
 *      将关联之后的数据写到CK
 */
public class DwsTradeSkuOrderWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 指定并行度
        env.setParallelism(4);

        //TODO 2.检查点相关的设置(略)

        //TODO 3.从kafka的下单主题中读取数据
        //3.1 声明消费的主题以及消费者组
        String topic = "dwd_trade_order_detail";
        String groupId = "dws_trade_sku_group";
        //3.2 创建消费者对象
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        //3.3 消费数据 封装为流
        DataStreamSource<String> kafkaStrDS = env.addSource(kafkaConsumer);
        //TODO 4.将读取的数据进行类型转换       jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(JSON::parseObject);
        // {"create_time":"2022-10-29 16:32:39","sku_num":"2","activity_rule_id":"5","split_original_amount":"138.0000",
        // "split_coupon_amount":"35.0","sku_id":"30","date_id":"2022-10-29","source_type_name":"用户查询",
        // "coupon_id":"2","user_id":"84","province_id":"14","source_type_code":"2401",
        // "row_op_ts":"2022-11-11 08:32:40Z","activity_id":"3","sku_name":"CAREMiLL","id":"351","order_id":"152",
        // "split_activity_amount":"14.0","split_total_amount":"89.0","ts":"1668155559"}
        // jsonObjDS.print(">>>>");

        //TODO 5.按照唯一键进行分组  下单业务过程最细粒度是明细 所以唯一键应该是订单明细的id
        KeyedStream<JSONObject, String> orderDetailIdKeyedDS = jsonObjDS.keyBy(jsonObj -> jsonObj.getString("id"));
        //TODO 6.使用Flink的状态 + 定时器完成去重
        SingleOutputStreamOperator<JSONObject> distinctDS = orderDetailIdKeyedDS.process(
            new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                private ValueState<JSONObject> lastValueState;

                @Override
                public void open(Configuration parameters) throws Exception {
                    ValueStateDescriptor<JSONObject> valueStateDescriptor = new ValueStateDescriptor<JSONObject>("lastValueState", JSONObject.class);
                    lastValueState = getRuntimeContext().getState(valueStateDescriptor);
                }

                @Override
                public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                    //获取状态的值
                    JSONObject lastValue = lastValueState.value();
                    if (lastValue == null) {
                        //如果状态中的内容为空，说明当前明细现在还没有数据过来，这条数据是第一条，我们将其放到状态中
                        lastValueState.update(jsonObj);
                        //同时注册5秒之后执行的定时器
                        long currentProcessingTime = ctx.timerService().currentProcessingTime();
                        ctx.timerService().registerProcessingTimeTimer(currentProcessingTime + 5000L);
                    } else {
                        //如果状态中的内容不为空，说明当前明细已经有数据过来,要用当前这条数据和状态中已经来的数据进行时间的比较
                        // "row_op_ts":"2022-11-11 08:32:40Z"
                        String lastRowOpTs = lastValue.getString("row_op_ts");
                        String curRowOpTs = jsonObj.getString("row_op_ts");
                        if (TimestampLtz3CompareUtil.compare(lastRowOpTs, curRowOpTs) <= 0) {
                            lastValueState.update(jsonObj);
                        }
                    }
                }

                @Override
                public void onTimer(long timestamp, OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
                    JSONObject lastValue = lastValueState.value();
                    if (lastValue != null) {
                        out.collect(lastValue);
                    }
                    lastValueState.clear();
                }
            }
        );

        //TODO 7.将jsonObj转换为实体类对象
        SingleOutputStreamOperator<TradeSkuOrderBean> orderBeanDS = distinctDS.map(
            new MapFunction<JSONObject, TradeSkuOrderBean>() {
                @Override
                public TradeSkuOrderBean map(JSONObject jsonObj) throws Exception {
                    String userId = jsonObj.getString("user_id");
                    String skuId = jsonObj.getString("sku_id");
                    String orderId = jsonObj.getString("order_id");
                    Double originalAmount = jsonObj.getDouble("split_original_amount");
                    Double couponAmount = jsonObj.getDouble("split_coupon_amount");
                    Double activityAmount = jsonObj.getDouble("split_activity_amount");
                    Double splitTotalAmount = jsonObj.getDouble("split_total_amount");
                    Long ts = jsonObj.getLong("ts")*1000;
                    TradeSkuOrderBean orderBean = TradeSkuOrderBean.builder()
                        .userId(userId)
                        .skuId(skuId)
                        .orderUuCount(0L)
                        .orderIdSet(new HashSet<String>(Collections.singleton(orderId)))
                        .originalAmount(originalAmount)
                        .couponAmount(couponAmount == null ? 0.0 : couponAmount)
                        .activityAmount(activityAmount == null ? 0.0 : activityAmount)
                        .orderAmount(splitTotalAmount)
                        .ts(ts)
                        .build();
                    return orderBean;
                }
            }
        );

        //TODO 8.按照user_id进行分组  以便计算下单独立用户
        KeyedStream<TradeSkuOrderBean, String> userIdKeyedDS = orderBeanDS.keyBy(TradeSkuOrderBean::getUserId);
        //TODO 9.使用Flink的状态编程 判断是否为下单独立用户
        SingleOutputStreamOperator<TradeSkuOrderBean> withUuDS = userIdKeyedDS.process(
            new KeyedProcessFunction<String, TradeSkuOrderBean, TradeSkuOrderBean>() {
                private ValueState<String> lastOrderDateState;

                @Override
                public void open(Configuration parameters) throws Exception {
                    ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<String>("lastOrderDateState", String.class);
                    valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                    lastOrderDateState = getRuntimeContext().getState(valueStateDescriptor);
                }

                @Override
                public void processElement(TradeSkuOrderBean orderBean, Context ctx, Collector<TradeSkuOrderBean> out) throws Exception {
                    //获取状态中的上次下单日期
                    String lastOrderDate = lastOrderDateState.value();
                    Long ts = orderBean.getTs();
                    String curOrderDate = DateFormatUtil.toDate(ts);
                    if (StringUtils.isEmpty(lastOrderDate) || !lastOrderDate.equals(curOrderDate)) {
                        orderBean.setOrderUuCount(1L);
                        lastOrderDateState.update(curOrderDate);
                    }
                    out.collect(orderBean);
                }
            }
        );
        // withUuDS.print(">>>");
        //TODO 10.指定Watermark以及提取事件时间字段
        SingleOutputStreamOperator<TradeSkuOrderBean> withWatermarkDS = withUuDS.assignTimestampsAndWatermarks(
            WatermarkStrategy
                .<TradeSkuOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(
                    new SerializableTimestampAssigner<TradeSkuOrderBean>() {
                        @Override
                        public long extractTimestamp(TradeSkuOrderBean orderBean, long recordTimestamp) {
                            return orderBean.getTs();
                        }
                    }
                )
        );
        //TODO 11.按照sku粒度进行分组
        KeyedStream<TradeSkuOrderBean, String> skuIdKeyedDS = withWatermarkDS.keyBy(TradeSkuOrderBean::getSkuId);
        //TODO 12.开窗
        WindowedStream<TradeSkuOrderBean, String, TimeWindow> windowDS
            = skuIdKeyedDS.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));
        //TODO 13.聚合计算
        SingleOutputStreamOperator<TradeSkuOrderBean> reduceDS = windowDS.reduce(
            new ReduceFunction<TradeSkuOrderBean>() {
                @Override
                public TradeSkuOrderBean reduce(TradeSkuOrderBean value1, TradeSkuOrderBean value2) throws Exception {
                    value1.getOrderIdSet().addAll(value2.getOrderIdSet());
                    value1.setOrderUuCount(value1.getOrderUuCount() + value2.getOrderUuCount());
                    value1.setOriginalAmount(value1.getOriginalAmount() + value2.getOriginalAmount());
                    value1.setActivityAmount(value1.getActivityAmount() + value2.getActivityAmount());
                    value1.setCouponAmount(value1.getCouponAmount() + value2.getCouponAmount());
                    value1.setOrderAmount(value1.getOrderAmount() + value2.getOrderAmount());
                    return value1;
                }
            },
            new WindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                @Override
                public void apply(String groupId, TimeWindow window, Iterable<TradeSkuOrderBean> input, Collector<TradeSkuOrderBean> out) throws Exception {
                    String stt = DateFormatUtil.toYmdHms(window.getStart());
                    String edt = DateFormatUtil.toYmdHms(window.getEnd());
                    for (TradeSkuOrderBean orderBean : input) {
                        orderBean.setStt(stt);
                        orderBean.setEdt(edt);
                        orderBean.setTs(System.currentTimeMillis());
                        orderBean.setOrderCount((long) orderBean.getOrderIdSet().size());
                        out.collect(orderBean);
                    }
                }
            }
        );
        // reduceDS.print(">>>");
        //TODO 14.关联sku维度
        /*
        //同步的方式，进行维度关联
        SingleOutputStreamOperator<TradeSkuOrderBean> skuInfoDS = reduceDS.map(
            new MapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
                @Override
                public TradeSkuOrderBean map(TradeSkuOrderBean orderBean) throws Exception {
                    //根据流中的对象获取要查询的维度主键
                    String skuId = orderBean.getSkuId();
                    //根据维度的主键到维度表中查询对应的维度数据
                    //ID,SPU_ID,PRICE,SKU_NAME,SKU_DESC,WEIGHT,TM_ID,CATEGORY3_ID,SKU_DEFAULT_IMG,IS_SALE,CREATE_TIME
                    JSONObject dimInfoJsonObj = DimUtil.getDimInfo("dim_sku_info", skuId);
                    //将相关的维度属性补充到流中的对象属性上
                    orderBean.setSkuName(dimInfoJsonObj.getString("SKU_NAME"));
                    orderBean.setSpuId(dimInfoJsonObj.getString("SPU_ID"));
                    orderBean.setTrademarkId(dimInfoJsonObj.getString("TM_ID"));
                    orderBean.setCategory3Id(dimInfoJsonObj.getString("CATEGORY3_ID"));

                    return orderBean;
                }
            }
        );
        skuInfoDS.print(">>>>");*/

        // 将异步 I/O 操作应用于 DataStream 作为 DataStream 的一次转换操作
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = AsyncDataStream.unorderedWait(
            reduceDS,
            //实现分发请求的AsyncFunction
            new DimAsyncFunction<TradeSkuOrderBean>("dim_sku_info") {
                @Override
                public void join(JSONObject dimInfoJsonObj, TradeSkuOrderBean orderBean) {
                    orderBean.setSkuName(dimInfoJsonObj.getString("SKU_NAME"));
                    orderBean.setSpuId(dimInfoJsonObj.getString("SPU_ID"));
                    orderBean.setTrademarkId(dimInfoJsonObj.getString("TM_ID"));
                    orderBean.setCategory3Id(dimInfoJsonObj.getString("CATEGORY3_ID"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getSkuId();
                }
            },
            60,
            TimeUnit.SECONDS
        );

        //TODO 15.关联spu维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withSpuInfoDS = AsyncDataStream.unorderedWait(
            withSkuInfoDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_spu_info") {
                @Override
                public void join(JSONObject dimInfoJsonObj, TradeSkuOrderBean orderBean) {
                    // id,spu_name,description,category3_id,tm_id
                    orderBean.setSpuName(dimInfoJsonObj.getString("SPU_NAME"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getSpuId();
                }
            },
            60, TimeUnit.SECONDS
        );
        // withSpuInfoDS.print(">>>>");

        //TODO 16.关联tm维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withTmDS = AsyncDataStream.unorderedWait(
            withSpuInfoDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_trademark") {
                @Override
                public void join(JSONObject dimInfoJsonObj, TradeSkuOrderBean tradeSkuOrderBean) {
                    // id,tm_name
                    tradeSkuOrderBean.setTrademarkName(dimInfoJsonObj.getString("TM_NAME"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getTrademarkId();
                }
            },
            60, TimeUnit.SECONDS
        );
        //TODO 17.关联category3维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withCategory3Stream = AsyncDataStream.unorderedWait(
            withTmDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category3".toUpperCase()) {
                @Override
                public void join(JSONObject jsonObj,TradeSkuOrderBean javaBean)  {
                    javaBean.setCategory3Name(jsonObj.getString("name".toUpperCase()));
                    javaBean.setCategory2Id(jsonObj.getString("category2_id".toUpperCase()));
                }

                @Override
                public String getKey(TradeSkuOrderBean javaBean) {
                    return javaBean.getCategory3Id();
                }
            },
            5 * 60, TimeUnit.SECONDS
        );

        //TODO 18.关联category2维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withCategory2Stream = AsyncDataStream.unorderedWait(
            withCategory3Stream,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category2".toUpperCase()) {
                @Override
                public void join(JSONObject jsonObj, TradeSkuOrderBean javaBean) {
                    javaBean.setCategory2Name(jsonObj.getString("name".toUpperCase()));
                    javaBean.setCategory1Id(jsonObj.getString("category1_id".toUpperCase()));
                }

                @Override
                public String getKey(TradeSkuOrderBean javaBean) {
                    return javaBean.getCategory2Id();
                }
            },
            5 * 60, TimeUnit.SECONDS
        );

        //TODO 19.关联category1维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withCategory1Stream = AsyncDataStream.unorderedWait(
            withCategory2Stream,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category1".toUpperCase()) {
                @Override
                public void join(JSONObject jsonObj,TradeSkuOrderBean javaBean) {
                    javaBean.setCategory1Name(jsonObj.getString("name".toUpperCase()));
                }

                @Override
                public String getKey(TradeSkuOrderBean javaBean) {
                    return javaBean.getCategory1Id();
                }
            },
            5 * 60, TimeUnit.SECONDS
        );

        //TODO 20.将关联的结果写到CK
        withCategory1Stream.print(">>>");
        withCategory1Stream.addSink(
            MyClickhouseUtil.getSinkFunction("insert into dws_trade_sku_order_window values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)")
        );
        env.execute();
    }
}
